package com.yujiahao.bigdata.rdd.transform

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
    TODO 转换算子--两个数据集的操作
      1、交集，并集，差集，拉链
      2、注意拉链两个数据集数量必须相同，泛型可以不一样西。
      3、交集，并集，差集--这几个泛型必须一样，数量可以不一样。
 */

object Spark_RDD_Pper_Transform_ManyCollections {
  def main(args: Array[String]): Unit = {


    //TODO 1、获得连接
    val conf: SparkConf = new SparkConf().setAppName("WordCount").setMaster("local")
    val sc: SparkContext = new SparkContext(conf)

    //TODO 3、业务逻辑 --测试两个数据集之间的交集、并集、差集、拉链
    val rdd1 = sc.makeRDD(
      List(
        1, 2, 3, 4
      ), 2
    )
    val rdd2 = sc.makeRDD(
      List(
        3, 4, 5, 6, 7, 8
      ), 3
    )
    val rdd3 = sc.makeRDD(
      List(
        "3", "4", "5", "6"
      ), 2
    )
    val haha = sc.makeRDD(
      List((1 -> "a"), (2 -> "b"), (3 -> "c"))
    )
    val haha1 = sc.makeRDD(List(
      (1, "d"), (2, "e"), (3, "f")
    ))

    println("测试两个键值对进行合并集合："+ haha.union(haha1).collect().mkString(" "))

    //TODO 1）交集--intersection
    println("测试交集：" + rdd1.intersection(rdd2).collect().mkString(" "))
    //注意交集两个集合的泛型要一致不然就报错：type mismatch;
    // rdd1.intersection(rdd3).collect().foreach(println)

    //TODO 2）并集--union
    println("测试并集：" + rdd1.union(rdd2).collect().mkString(" "))
    //注意交集两个集合的泛型要一致不然就报错：type mismatch;
    //println("测试并集：" + rdd1.union(rdd3).collect().mkString(" "))

    //TODO 3）差集--subtract
    // 注意交集两个集合的泛型要一致不然就报错：type mismatch;
    println("测试差集：" + rdd1.subtract(rdd2).collect().mkString(" "))

    //TODO 4）拉链--zip

    //TODO 注意：Spark中两个数据集中数据拉链必须相等：不然报错
    // TODO Can't zip RDDs with unequal numbers of partitions: List(2, 3)
    //println("测试拉链：" + rdd1.zip(rdd2).collect().mkString(" "))
    //TODO 拉链类型可以不一样，但是一定要保证数量对的上
    println("测试拉链：" + rdd1.zip(rdd3).collect().mkString(" "))


    //TODO 2、关闭连接
    sc.stop()
  }
}